package cep;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;
import java.util.Objects;

public class TestCep02 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> stringDataStreamSource = env.readTextFile("data/cep01.txt");
        SingleOutputStreamOperator<UserRegist> ds = stringDataStreamSource.map(new MapFunction<String, UserRegist>() {
            int i=0;
            @Override
            public UserRegist map(String value) throws Exception {
                i++;
                String[] split = value.split(" ");
                if(i%2==0){
                    return new UserRegist(Integer.valueOf(split[0]), split[1], split[2]+":"+i);
                }
                //奇数返回子类
                SubClass subClass = new SubClass();
                subClass.setId(Integer.valueOf(split[0]));
                subClass.setName(split[1]);
                subClass.setAction(split[2]+":"+i);
                return subClass;
            }
        });

        //过滤出是  SubClass这个子类
        Pattern<UserRegist, SubClass> pattern = Pattern.<UserRegist>begin("begin").subtype(SubClass.class);

        PatternStream<UserRegist> ds1 = CEP.pattern(ds, pattern);

        ds1.select(new PatternSelectFunction<UserRegist, String>() {
            @Override
            public String select(Map<String, List<UserRegist>> map) throws Exception {
                map.forEach((k,v)->{
                    System.out.println(k);
                    v.forEach(t-> System.out.println(t));
                });
                //get阶段 get元素 getName
                return map.get("begin").get(0).getName();
            }
        }).print();

        env.execute();
    }

    public static class SubClass extends UserRegist{ }

    public static class UserRegist{
        private int id;
        private String name;
        private String action;

        public UserRegist() {
        }

        public UserRegist(int id, String name, String action) {
            this.id = id;
            this.name = name;
            this.action = action;
        }

        @Override
        public String toString() {
            return "UserRegist{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    ", action='" + action + '\'' +
                    '}';
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (!(o instanceof UserRegist)) return false;
            UserRegist that = (UserRegist) o;
            return getId() == that.getId() &&
                    Objects.equals(getName(), that.getName()) &&
                    Objects.equals(getAction(), that.getAction());
        }

        @Override
        public int hashCode() {

            return Objects.hash(getId(), getName(), getAction());
        }

        public int getId() {
            return id;
        }

        public void setId(int id) {
            this.id = id;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public String getAction() {
            return action;
        }

        public void setAction(String action) {
            this.action = action;
        }
    }
}

